package com.zyx.flinkdemo.sql.window.sql.tvf;

import com.zyx.flinkdemo.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author zyx
 * @since 2021/5/30 08:15
 * desc: 基于 事件时间 查询滑动TVF表 案例
 *  https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/dev/table/sql/queries/window-tvf/
 */
public class HopTvfProcessTimeDemo {
    public static void main(String[] args) throws Exception {
        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 2.读取端口数据创建流并转换每一行数据为JavaBean对象
        SingleOutputStreamOperator<WaterSensor> waterSensorDs = env
                .socketTextStream("linux201", 7777)
                .map(line -> {
                    String[] split = line.split(",");
                    return new WaterSensor(split[0],
                            Long.parseLong(split[1]),
                            Integer.parseInt(split[2]));
                });

        // 3.将流转换为表并指定处理时间
        Table table = tableEnv.fromDataStream(waterSensorDs,
                $("id"),
                $("ts"),
                $("vc"),
                $("pt").proctime());
        tableEnv.createTemporaryView("water_sensor", table);

        // 4.SQL 实现滚动时间窗口
        Table result = tableEnv.sqlQuery("select id, window_start, window_end, count(id) as `count` from " +
                /* 可以将 TVF 作为一张表进行查询, 这种动态的 TVF 表实际上是老版本GroupWindow的代替和继承
                    Windowing TVFs is a replacement of legacy Grouped Window Functions.
                   可以使用 TUMBLE(TABLE data, DESCRIPTOR(timecol), size)
                    HOP(TABLE data, DESCRIPTOR(timecol), slide, size [, offset ])
                   而且 滑动窗口的大小应该是 滑动时间 的整数倍!!! 否则将会抛出以下异常
                    HOP table function based aggregate requires size must be an integral multiple of slide */
                "TABLE(HOP(TABLE water_sensor, DESCRIPTOR(pt), INTERVAL '3' SECONDS, INTERVAL '3' SECONDS)) " +
                // 也可以使用 Named Param指定, 但是 DATA参数需要放置在首位!
                /*
                   TABLE(HOP(DATA => TABLE water_sensor,
                        TIMECOL => DESCRIPTOR(pt),
                        SLIDE => INTERVAL '3' SECONDS,
                        SIZE => INTERVAL '3' SECONDS))
                 */
                "group by id, window_start, window_end"
        );

        // 5.将结果表转换为流进行输出
        // tableEnv.toAppendStream(result, Row.class).print()

        // 5.直接输出表
        result.execute().print();

        //6.执行任务
        env.execute();
    }
}
